package com.atguigu.flink.state;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * Created by 黄凯 on 2023/6/21 0021 16:41
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * TTL:  time to live  存活时间。
 * <p>
 * 状态默认都是在内存中存储。不清空，不删除。
 */
public class Flink11_StateTTL {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
                Builder(Time.seconds(15)): 当前状态的存活时间配置是15s(物理时间)

                如何判断状态过期?
                    如果状态没有启用ttl，就是永不过期。状态在存储时是K-V
                        S1: ListState<Integer>
                    如果状态启用ttl，是会过期。状态在存储时是K-V
                        S1: ListState<Integer>, accessTime= 上次被访问的时间。


                    当前尝试去读|写状态，先获取到状态的accessTime，使用 当前时间 -  accessTime = 距离上一次被访问又经过的时间t

                    t > 15s，说明状态已经15s没有被访问，是一个过期的状态！


                accessTime如何被更新?
                    取决于策略。
                        setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        当前的状态，被读或写的时候，会更新accessTime。


                当前的状态，被创建后，只被访问一次，后续再没有被访问，如何判断它过期，删除？
                    flink内部会维护一个线程，线程会不断扫描每个状态的accessTime，如果扫描时发现已经超时过期了，会将状态删除。

         */

        StateTtlConfig stateTtlConfig = new StateTtlConfig.Builder(Time.seconds(15))
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                //如果状态已经过期了，读的时候就不返回，返回null
                .neverReturnExpired()
                .build();

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {

                    private ListState<Integer> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ListStateDescriptor<Integer> listStateDescriptor = new ListStateDescriptor<>("top3", Integer.class);

                        //当前状态启用了TTL
                        listStateDescriptor.enableTimeToLive(stateTtlConfig);

                        //找到状态的备份存储,取出状态并赋值
                        listState = getRuntimeContext().getListState(listStateDescriptor);

                    }

                    @Override
                    public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {

                        //加入集合
                        listState.add(value.getVc());

                        //排序比较取前3。 Stream
                        List<Integer> top3 = StreamSupport.stream(listState.get().spliterator(), true)
                                .sorted(Comparator.reverseOrder())
                                .limit(3)
                                .collect(Collectors.toList());

                        out.collect(ctx.getCurrentKey() + " 目前的前3vc是:" + top3);

                        //始终只留当前的前三名
                        listState.update(top3);

                    }
                }
        ).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
